﻿/**
 * @file thread_pool.h
 * @date 2021/10/28 14:10
 * 
 * @author harrybao (harrybao@123u.com)
 * 
 * @brief 
 * 
 * @version v1.0.0
 * 
 * @copyright Copyright © 2021 harrybao All Rights Reserved
 * 
 * @par history:
 * <table>
 * <tr><th>Date                     <th>Version     <th>Author      <th>Description
 * <tr><td>2021/10/28 14:10         <td>1.0.0       <td>harrybao    <td>创建初始版本
 * </table>
 */

#pragma once
#include <functional>
#include <deque>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <memory>
#include <cassert>
#include <chrono>
#include "common/define/noncopyable.h"

/**
 * @brief 线程池
 */
class thread_pool : public noncopyable
{
public:
    using task_func = std::function<void()>;

    thread_pool()
        : m_running(false)
    {
    }
    thread_pool(const std::string &pool_name) {}
    
    ~thread_pool()
    {
        if (m_running)
        {
            stop();
        }
    }
    
    const std::string& name() const
    { return m_pool_name; }

    void start(const uint32_t thread_num)
    {
        assert(m_threads.empty() && !m_running);
        m_running = true;
        m_threads.reserve(thread_num);
        for (uint32_t i = 0; i < thread_num; ++i)
        {
            m_threads.emplace_back(new std::thread(&thread_pool::process, this, i));
        }
    }

    void stop()
    {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_running = false;
            m_not_empty_cond.notify_all();
        }

        for (auto &th : m_threads)
        {
            th->join();
        }

        m_threads.clear();

        // TODO 直接跑完全部任务
        task_func task;
        while (!m_task_que.empty())
        {
            task = m_task_que.front();
            m_task_que.pop_front();
            if (task)
            {
                task();
            }
        }

    }

    void add_task(const task_func &task)
    {
        if (m_threads.empty())
        {
            task();
        }
        else
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_not_full_cond.wait(lock, [this]() -> bool {
                return !this->is_full();
            });

            m_task_que.push_back(task);
            m_not_empty_cond.notify_one();
        }
    }
private:
    uint32_t seq_id_{ 0 };
public:
    void AddValue()
    {
        //std::cout << "AddValue" << " thread_id=" << 1 << std::endl;
        std::unique_lock<std::mutex> lock(m_mutex);
        list_count_++;
        seq_id_++;
        //std::cout << "+1 cout=" << list_count_ << std::endl;
    }
    void RemoveValue()
    {
        //std::cout << "RemoveValue" << " thread_id=" << 2 << std::endl;
        std::unique_lock<std::mutex> lock(m_mutex);
        if (list_count_ > 0)
        {
            list_count_--;
            //std::cout << "-1 cout=" << list_count_ << std::endl;
        }
    }
    bool TestValue(uint32_t nId)
    {
        //std::unique_lock<std::mutex> lock(m_mutex);
        if (list_count_ == 0)
        {
            //std::cout << "0 cout=" << list_count_ << std::endl;
            if (list_count_ != 0)
            {
                std::cout << "sleep 20" << " count=" << list_count_ 
                    << " thread_id=" << nId
                    << " seq_id=" << seq_id_
                    << std::endl;
            }
            //std::this_thread::sleep_for(std::chrono::milliseconds(20));
            return true;
        }
        return false;
    }
private:
    void process(uint32_t idx)
    {
        //auto tp = std::chrono::system_clock::now();
        //std::cout << std::chrono::system_clock::to_time_t(tp) << std::endl;
        //std::this_thread::sleep_for(std::chrono::milliseconds(2000));
        //printf("current thread id: %d\n", nId);
        try
        {
            while (m_running)
            {
                //std::this_thread::sleep_for(std::chrono::milliseconds(5));
                task_func task = take();
                if (task)
                {
                    task();
                }
            }
        }
        catch (...)
        {
            // TODO
            throw;
        }
    }

    task_func take()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_not_empty_cond.wait(lock, [this]() -> bool {
            return !this->m_task_que.empty() || !this->m_running;
        });

        task_func task;
        if (!m_task_que.empty())
        {
            task = m_task_que.front();
            m_task_que.pop_front();
            if (m_max_task_que_size > 0)
            {
                m_not_full_cond.notify_one();
            }
        }
        return task;
    }

    bool is_full()
    {
        return m_max_task_que_size > 0 && m_task_que.size() >= m_max_task_que_size;
    }
private:
    mutable std::mutex m_mutex;
    std::condition_variable m_not_empty_cond;
    std::condition_variable m_not_full_cond;
    std::vector<std::unique_ptr<std::thread>> m_threads;
    std::deque<task_func> m_task_que;
    size_t m_max_task_que_size;
    bool m_running{ false };
    volatile uint32_t list_count_{ 0 };
    std::string m_pool_name;
};